package com.atguigu.wuliu.app.dws;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.wuliu.app.func.BeanToJsonStrFunction;
import com.atguigu.wuliu.app.func.DimAsyncByForeignKeyFunction;
import com.atguigu.wuliu.app.func.MyTriggerFunction;
import com.atguigu.wuliu.bean.DwdBoundfenJianBean;
import com.atguigu.wuliu.bean.DwdTransReceiveDetailBean;
import com.atguigu.wuliu.bean.DwsBoundOrgSortDayBean;
import com.atguigu.wuliu.bean.DwsTransOrgReceiveDayBean;
import com.atguigu.wuliu.utils.DateFormatUtil;
import com.atguigu.wuliu.utils.DorisUtil;
import com.atguigu.wuliu.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/*中转域机构粒度分拣汇总表*/
public class DwsBoundOrgSortDay {
    public static void main(String[] args) throws Exception {
        //1.获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
       // System.setProperty("HADOOP_USER_NAME","atguigu");
        //2.设置检查点
        env.enableCheckpointing(5000L);
        //3.从kafka读取揽收表数据：tms_dwd_trans_receive_detail
        String topic = "tms_dwd_bound_sort";
        String group = "dwd_bound_sort_group";
        DataStreamSource<String> kafakSourceDS = env.fromSource(KafkaUtil.getKafkaSource(topic, group), WatermarkStrategy.noWatermarks(), "kafakSource");
        //kafakSourceDS.print("原始数据");

        // {"id":"19043","orderId":"11441","orgId":"5","sortTime":"2023-08-18 07:08:07","sorterEmpId":"205","ts":1692313687}

        //4.数据转为自定义实体类 DwsBoundOrgSortDayBean
        SingleOutputStreamOperator<DwsBoundOrgSortDayBean> mapDS = kafakSourceDS.map(
                new MapFunction<String, DwsBoundOrgSortDayBean>() {
                    @Override
                    public DwsBoundOrgSortDayBean map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        Long ts = jsonObject.getLong("ts");
                        String orgId = jsonObject.getString("orgId");

                        DwsBoundOrgSortDayBean dwsBoundOrgSortDayBean = DwsBoundOrgSortDayBean.builder()
                                .ts(ts)
                                .orgId(orgId)
                                .sortCountBase(1L)
                                .build();
                        return dwsBoundOrgSortDayBean;
                    }
                }
        );
        mapDS.print("转换为实体类");



        //6.设置watermaerks提取事件事件字段ts
        SingleOutputStreamOperator<DwsBoundOrgSortDayBean> withWatermarkDS = mapDS.assignTimestampsAndWatermarks(
                // WatermarkStrategy.<DwsBoundOrgSortDayBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                WatermarkStrategy
                        .<DwsBoundOrgSortDayBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<DwsBoundOrgSortDayBean>() {
                                    @Override
                                    public long extractTimestamp(DwsBoundOrgSortDayBean boundOrgSortDayBean, long recordTimestamp) {
                                        return boundOrgSortDayBean.getTs();
                                    }
                                }
                        )
        );
        //TODO 5.按照机构id进行分组
        KeyedStream<DwsBoundOrgSortDayBean, String> keyedDS = withWatermarkDS.keyBy(DwsBoundOrgSortDayBean::getOrgId);

        //TODO 6.开窗
        WindowedStream<DwsBoundOrgSortDayBean, String, TimeWindow> windowDS
                = keyedDS.window(TumblingEventTimeWindows.of(Time.days(1L)));

        //TODO 7.指定自定义的触发器
        WindowedStream<DwsBoundOrgSortDayBean, String, TimeWindow> triggerDS
                = windowDS.trigger(new MyTriggerFunction<DwsBoundOrgSortDayBean>());

        //TODO 8.聚合
        SingleOutputStreamOperator<DwsBoundOrgSortDayBean> reduceDS = windowDS.reduce(
                new ReduceFunction<DwsBoundOrgSortDayBean>() {
                    @Override
                    public DwsBoundOrgSortDayBean reduce(DwsBoundOrgSortDayBean bean, DwsBoundOrgSortDayBean lastBean) throws Exception {
                      bean.setSortCountBase(bean.getSortCountBase()+ lastBean.getSortCountBase());
                        return bean;
                    }
                }
        );

        reduceDS.print("聚合后数据");
    /*    //补充维度
        SingleOutputStreamOperator<DwsTransOrgReceiveDayBean> withOrgNameStream = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncByForeignKeyFunction<DwsTransOrgReceiveDayBean>("dim_base_organ") {
                    @Override
                    public void join(DwsTransOrgReceiveDayBean bean, JSONObject dimJsonObj) {
                        bean.setOrgName(dimJsonObj.getString("org_name"));
                    }

                    @Override
                    public Tuple2<String, String> getForeignKey(DwsTransOrgReceiveDayBean obj) {
                        return Tuple2.of("id",obj.getOrgId());
                    }


                },
                60, TimeUnit.SECONDS
        );

        // 11.2 补充城市名称
        SingleOutputStreamOperator<DwsTransOrgReceiveDayBean> withCityNameStream = AsyncDataStream.unorderedWait(
                withOrgNameStream,
                new DimAsyncByForeignKeyFunction<DwsTransOrgReceiveDayBean>("dim_base_region_info") {
                    @Override
                    public void join(DwsTransOrgReceiveDayBean bean, JSONObject dimJsonObj) {
                        bean.setCityName(dimJsonObj.getString("name"));
                    }

                    @Override
                    public Tuple2<String, String> getForeignKey(DwsTransOrgReceiveDayBean bean) {
                        return Tuple2.of("id",bean.getCityId());
                    }


                },
                60, TimeUnit.SECONDS
        );

        // 11.3 补充省份名称
        SingleOutputStreamOperator<DwsTransOrgReceiveDayBean> fullStream = AsyncDataStream.unorderedWait(
                withCityNameStream,
                new DimAsyncByForeignKeyFunction<DwsTransOrgReceiveDayBean>("dim_base_region_info") {
                    @Override
                    public void join(DwsTransOrgReceiveDayBean bean, JSONObject dimJsonObj) {
                        bean.setProvinceName(dimJsonObj.getString("name"));
                    }

                    @Override
                    public Tuple2<String,String> getForeignKey(DwsTransOrgReceiveDayBean bean) {
                        return Tuple2.of("id",bean.getProvinceId());
                    }
                },
                60, TimeUnit.SECONDS
        );
        fullStream.print("最终结果");*/


    /*   // TODO 12. 写出到 doris
        fullStream.map(
                new BeanToJsonStrFunction<>()
        ).sinkTo(DorisUtil.getDorisSink("dws_trans_org_receive_day_base"));
*/


        env.execute();
    }
}
